package com.zhu.utils;

import com.zhu.config.ClusterParametersConfig;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.nio.charset.StandardCharsets;
import java.util.Properties;

public class ZhuKafkaUtil {

    public static FlinkKafkaConsumer<String> getFlinkKafkaConsumer(String topic, String groupId) {

        Properties properties = new Properties();
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, ClusterParametersConfig.KAFKA_BOOTSTRAP_SERVER);

        return new FlinkKafkaConsumer<String>(
                topic,
                new KafkaDeserializationSchema<String>() {
                    /**
                     * 设置是否是无解流的重点
                     * @param s 无界流的重点
                     * @return
                     */
                    @Override
                    public boolean isEndOfStream(String s) {
                        //无界流 false
                        return false;
                    }

                    /**
                     * 自带的序列化格式不允许字符传为空 会直接抛出异常  这里自定义
                     * @param consumerRecord  包装kafka中的每一条数据
                     * @return
                     * @throws Exception
                     */
                    @Override
                    public String deserialize(ConsumerRecord<byte[], byte[]> consumerRecord) throws Exception {
                        if (consumerRecord == null || consumerRecord.value() == null) {
                            return null;  //SimpleStringSchema 不允许数据为空 会返回异常
                        } else {
                            return new String(consumerRecord.value()); //正常数据写出
                        }
                    }

                    @Override
                    public TypeInformation<String> getProducedType() {
                        return BasicTypeInfo.STRING_TYPE_INFO;   //参考 SimpleStringSchema
                    }
                }, properties
        );
    }

    public static FlinkKafkaProducer<String> getFlinkKafkaProducer(String topic) {

        return new FlinkKafkaProducer<String>(ClusterParametersConfig.KAFKA_BOOTSTRAP_SERVER
                , topic
                , new SimpleStringSchema());
    }

    public static FlinkKafkaProducer<String> getFlinkKafkaProducer(String topic, String defaultTopic) {

        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,ClusterParametersConfig.KAFKA_BOOTSTRAP_SERVER);
        return new FlinkKafkaProducer<String>(
                defaultTopic, new KafkaSerializationSchema<String>() {
            @Override
            public ProducerRecord<byte[], byte[]> serialize(String value, @Nullable Long aLong) {
                if(value == null) {
                    return new ProducerRecord<>(topic,"".getBytes(StandardCharsets.UTF_8));
                }
                return new ProducerRecord<>(topic,value.getBytes(StandardCharsets.UTF_8));
            }
        },properties, FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
    }

    //getKafkaDDL 连接到kafka的
    public static String getKafkaDDL(String topic,String groupId) {
        return " with ( 'connector' = 'kafka', " +
                "'topic' = '"+ topic +"', " +
                "'properties.bootstrap.servers' = 'hadoop102:9092' ," +
                "'properties.group.id' = '"+ groupId + "' ," +
                "'format' = 'json' ," +
                "'scan.startup.mode' = 'group-offsets' )";
    }

    //getKafkaDDL  输出到kafka的表

    /**
     *
     * 输出到kafka
     * @param topic 输出到kafka的目标主题
     * @return   DDL
     */
    public static String getKafkaSinkDDL(String topic) {
        return "with ( 'connector' = 'kafka' ," +
                "'topic' = '"+ topic + "' ," +
                "'properties.bootstrap.servers' = 'hadoop102:9092' ," +
                "'format' = 'json' " +
                ")";
    }

    /**
     * upsert 针对撤回流数据
     * @param topic  kafka主题
     * @return
     */
    public static String getKafkaUpsertSinkDDL(String topic){
        return "with (" +
                "'connector' = 'upsert-kafka', " +
                "'topic' = '" + topic + "', " +
                "'properties.bootstrap.servers' = 'hadoop102:9092', " +
                "'key.format' = 'json', " +
                "'value.format' = 'json' " +
                ")";
    }

    //获取ods base db 数据

    /**
     * kafka source topic_db 主题的数据
     * @param groupId
     * @return  拼接好 kafka数据源的DDL
     */
    public static String getTopicDB(String groupId) {
        return  "create table topic_db(" +
                "`database` string, " +
                "`table` string, " +
                "`type` string, " +  //upsert
                "`data` map<string, string>, " +
                "`old` map<string, string>, " +
                "`pt` as PROCTIME(), " +   //maxwell的事件时间 没有必要
                "`ts` string " +
                ")" + getKafkaDDL("topic_db",groupId);
    }
}
